package com.ctsi.ftp.util;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.SocketException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTP;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.Path;


public class FTPUtil {
	private static Log log = LogFactory.getLog(FTPUtil.class);
	//SXS:change to 10000 gz文件行数**********
	private static int lineLimit=Integer.valueOf(ConfigProperties.getProperty("lineLimit"));
	//SXS:文件类型索引
	private static int fileTypeIndex=Integer.valueOf(ConfigProperties.getProperty("fileTypeIndex"));

    static int number = 0;
	public  static boolean uploadFileByDirectory(String ftpUrl, int ftpPort,
			String ftpUserName, String ftpPassword, String remotePath,
			Path[] files, String successHdfsPath, String filePostfix,
			boolean isGZip) {
			// TODO Auto-generated method stub
			//初始化，表示上传失败
			boolean success = false;
			//创建FTPClient对象
			FTPClient ftp=new FTPClient();

			//连接FTP服务器
			try {
				try {
					ftp.connect(ftpUrl,ftpPort);
				} catch (SocketException e) {
					e.printStackTrace();
					throw new Exception("无法连接FTP服务器：["+ftpUrl+":"+ftpPort+"]");
				} catch (IOException e) {
					e.printStackTrace();
					throw new Exception("无法连接FTP服务器：["+ftpUrl+":"+ftpPort+"]");
				}

				//登录ftp
				try {
					ftp.login(ftpUserName, ftpPassword);
				} catch (IOException e) {
					e.printStackTrace();
					throw new Exception("无法使用用户名["+ftpUserName+"]密码["+ftpPassword+"]登录FTP服务器：["+ftpUrl+":"+ftpPort+"]");
				}
				//获取登录的返回值，如果是230，则表示登录成功
				int reply = ftp.getReplyCode();
				ftp.enterLocalPassiveMode();
				//判断返回登录成功，以2开头的返回值，会返回真
				if(!FTPReply.isPositiveCompletion(reply)){
					//失败，释放连接
					try {
						ftp.disconnect();
					} catch (IOException e) {
						e.printStackTrace();
						throw new Exception("无法释放与FTP服务器：["+ftpUrl+":"+ftpPort+"]的连接");
					}
					return success;
				}
				ftp.setSoTimeout(60*1000);
				//设置上传格式为二进制
				ftp.setFileType(FTP.BINARY_FILE_TYPE);
				//转到指定的上传目录
				try {
					boolean ret = ftp.changeWorkingDirectory(remotePath);
					if(!ret){
						throw new Exception("无法获取FTP服务器的目录：["+remotePath+"]");
					}
				} catch (IOException e) {
					e.printStackTrace();
					throw new Exception("无法获取FTP服务器的目录：["+remotePath+"]");
				}
					//将上传文件存储到指定目录
				String fileName = "";
				String gzFileName="";
				try {
					Path newFilePath = new Path(successHdfsPath);
					if((!HdfsUtil.existsHdfsPath(newFilePath))||(!HdfsUtil.isHdfsDirectory(newFilePath))){
						//如果,待放上传成功的文件的目录不存在,则创建一个新目录
						HdfsUtil.makeHdfsDir(newFilePath);
					}

					for( Path file:files){
						if (file==null){
							continue;
						}
						if (HdfsUtil.isHdfsFile(file)) {
							// 遍历待上传的文件
							fileName = file.getName();
							if(filePostfix!=null){
								if(!filePostfix.equals(".*")){
									if(!fileName.endsWith(filePostfix)){
										log.info("==============Not a specified file ========"+fileName);
										return true;
									}
								}
							}
							// 上传文件
							log.debug("==============begin upload file ========"+fileName);
							if(isGZip){
//								将文件拆分
								//按行读取输入文件
								FSDataInputStream in=HdfsUtil.ReadHDFSFileStream(file);
								BufferedReader lineRead=new BufferedReader(new InputStreamReader(in,"UTF-8"));
								/*源文件名即gz文件名
								//输出gz文件名
								//String fileNamePre=fileName.substring(0,fileName.lastIndexOf(".txt"));
								//String[] fileParent=file.getParent().toString().split(File.separator);
								*/
								/*通过路径获取gz文件名
								//获取当前文件路径
								/*String[] fileParent=file.getParent().toString().split("/");
								//通过路径获得文件类型
								String fileType=fileParent[5].substring(1);
								String fileNamePre=fileType+"_"+fileParent[6]+fileParent[7]+fileParent[8];
								*/
								//通过路径获得文件类型，判断文件输出类型（LUA或NUA）
								String[] fileParent=file.getParent().toString().split("/");
								//String fileType=fileParent[5].substring(1);
								//SXS-2016-04-13:文件类型在路径中的位置改为可配置
								String fileType=fileParent[fileTypeIndex];
								String fileNamePre="";
								if(fileType.contains("NetPub")){
									fileNamePre="LUA";
								}else{
									fileNamePre="NUA";
								}
								//SXS:change to catch the gzfilename in exception***********
								//String gzFileName=null;
								//信息ID号：由SMCG自动生成，使用固定12字符长度，从1开始，采取自增方式，超过12位后可以重复开始；
								//int gzCount=0;

								String line="";
                                int temp =0;
                                OutputStream out = null;

                                SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
                                //gzFileName=fileNamePre +"_"+ fileType + "_" + gzCount + "_"+ System.currentTimeMillis() + ".xml.gz";
                                //gzFileName=fileNamePre +"_"+ fileType + "_" + FileIdGenerator.getId() + "_"+ sdf.format(new Date()) + ".xml.gz";
								//SXS:the real final filename
								gzFileName=fileNamePre +"_" + FileIdGenerator.getId() + "_"+ sdf.format(new Date()) + ".xml.gz";
                                //System.out.println("1---FTPUtil...Uploading the gzFile:"+gzFileName);
                                while((line=lineRead.readLine())!=null){
                                    temp++;
                                    line+="\r\n";
                                    if(temp ==1){
                                    	String head="<?xml version=\"1.0\" encoding=\"UTF-8\"?>\r\n"
                                    			+ "<info id=\"\" type=\"nat_cdr_info\" resultnum=\"\">\r\n";
                                        String firstLine=head+line;
                                        out = uploadGzFileStream_head(ftp, gzFileName, firstLine.getBytes());
                                    }else if(temp==lineLimit){
                                        line+="</info>\r\n";
                                        uploadGzFileStream_tail(out, ftp, gzFileName, line.getBytes());
                                        //SXS:Use ID instead*****************
//										FileIdGenerator.nextId();
//                                      gzFileName=fileNamePre +"_"+ fileType + "_" + FileIdGenerator.getId() + "_"+ sdf.format(new Date()) + ".xml.gz";
										//SXS:the real final filename
										gzFileName=fileNamePre +"_" + FileIdGenerator.getId() + "_"+ sdf.format(new Date()) + ".xml.gz";
										temp=0;
                                        out = null;
										//**************debug*******************************************
										//System.out.println("2---FTPUtil...Uploading the gzFile:"+gzFileName);
                                    }else{
                                        appendGzFileStream(out, ftp,gzFileName, line.getBytes());
                                    }

                                }
								//关闭输入流
                                if(temp!=0) {
									line="</info>\r\n";
	                                uploadGzFileStream_tail(out, ftp, gzFileName, line.getBytes());
	                                lineRead.close();
	                                in.close();
	                                in = null;
									//SXS:add to solve null pointer exception**********************
									out=null;

                                }

							}
						}

						// 将上传成功的文件转移到新目录中
						log.debug("==============begin move file ========"+fileName);
						HdfsUtil.renameHdfsFile(file,new Path(successHdfsPath+"/"+file.getName()));
						log.debug("==============end move file ========"+fileName);
						success = true;
					}
				} catch (Exception e) {
					e.printStackTrace();
					throw new Exception("上传文件["+fileName+"|"+gzFileName+"]中的文件出错!");
				}

			} catch (Exception e1) {
				e1.printStackTrace();
			} finally{
				try {
					if(ftp.isConnected()){
						ftp.logout();
						ftp.disconnect();
					}
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			//退出FTP
			return success;
	}

	private static OutputStream uploadGzFileStream_head(FTPClient ftp,String newName,byte[] buf) throws Exception{
		// TODO Auto-generated method stub
		OutputStream out;
		out = ftp.storeFileStream(newName);
		byte[] gzBuf=GzUtil.gZip(buf, buf.length);
		out.write(gzBuf, 0, gzBuf.length);
//		out.close();
//    	// Must call completePendingCommand() to finish command.
//    	if(!ftp.completePendingCommand()) {
//    	    ftp.logout();
//    	    ftp.disconnect();
//    	    throw new Exception("上传文件["+newName+"]中的文件出错!");
//    	}
        return out;
	}
    private static void uploadGzFileStream_tail(OutputStream out, FTPClient ftp,String newName,byte[] buf) throws Exception{
        // TODO Auto-generated method stub
//        OutputStream out;
//        out = ftp.storeFileStream(newName);
        byte[] gzBuf=GzUtil.gZip(buf, buf.length);
        out.write(gzBuf, 0, gzBuf.length);
        out.close();
        // Must call completePendingCommand() to finish command.
        if(!ftp.completePendingCommand()) {
            ftp.logout();
            ftp.disconnect();
            throw new Exception("上传文件["+newName+"]中的文件出错!");
        }
    }
	public static void appendGzFileStream(OutputStream out, FTPClient ftp,String newName,byte[] buf) throws Exception{
//    	OutputStream out;
//		out = ftp.appendFileStream(newName);

		byte[] gzBuffer=GzUtil.gZip(buf,buf.length);
		out.write(gzBuffer, 0, gzBuffer.length);
//	 	out.close();
//    	 // Must call completePendingCommand() to finish command.
//    	 if(!ftp.completePendingCommand()) {
//    	     ftp.logout();
//    	     ftp.disconnect();
//    	     throw new Exception("上传文件["+newName+"]中的文件出错!");
//    	 }
	}

	public  static boolean uploadFiles(String ftpUrl, int ftpPort, String ftpUserName, String ftpPassword,
									   String remotePath, Path[] files, String successHdfsPath, String filePostfix,
									   boolean isGZip) {
		// TODO Auto-generated method stub
		//初始化，表示上传失败
		boolean success = false;
		//创建FTPClient对象
		FTPClient ftp=new FTPClient();

		//连接FTP服务器
		try {
			try {
				ftp.connect(ftpUrl, ftpPort);
			} catch (SocketException e) {
				e.printStackTrace();
				log.error("无法连接FTP服务器：[" + ftpUrl + ":" + ftpPort + "]");
				throw new Exception("无法连接FTP服务器：["+ftpUrl+":"+ftpPort+"]");
			} catch (IOException e) {
				e.printStackTrace();
				log.error("无法连接FTP服务器：[" + ftpUrl + ":" + ftpPort + "]");
				throw new Exception("无法连接FTP服务器：["+ftpUrl+":"+ftpPort+"]");
			}

			//登录ftp
			try {
				ftp.login(ftpUserName, ftpPassword);
			} catch (IOException e) {
				e.printStackTrace();
				log.error("无法使用用户名[" + ftpUserName + "]密码[" + ftpPassword + "]登录FTP服务器：[" + ftpUrl + ":" + ftpPort + "]");
				throw new Exception("无法使用用户名["+ftpUserName+"]密码["+ftpPassword+"]登录FTP服务器：["+ftpUrl+":"+ftpPort+"]");
			}
			//获取登录的返回值，如果是230，则表示登录成功
			int reply = ftp.getReplyCode();
			ftp.enterLocalPassiveMode();

			//判断返回登录成功，以2开头的返回值，会返回真
			if(!FTPReply.isPositiveCompletion(reply)){
				//失败，释放连接
				try {
					ftp.disconnect();
				} catch (IOException e) {
					e.printStackTrace();
					log.error("无法释放与FTP服务器：["+ftpUrl+":"+ftpPort+"]的连接");
					throw new Exception("无法释放与FTP服务器：["+ftpUrl+":"+ftpPort+"]的连接");
				}
				return success;
			}
			ftp.setSoTimeout(60*1000);
			//设置上传格式为二进制
			ftp.setFileType(FTP.BINARY_FILE_TYPE);
			//转到指定的上传目录
			try {
				boolean ret = ftp.changeWorkingDirectory(remotePath);
				if (!ret) {
					throw new Exception("无法获取FTP服务器的目录：["+remotePath+"]"+ret);
				}
			} catch (IOException e) {
				e.printStackTrace();
				log.error("无法获取FTP服务器的目录：[" + remotePath + "]");
				throw new Exception("无法获取FTP服务器的目录：["+remotePath+"]");
			}
			//debug************************************************
			/*FTPFile[] ftpDir=ftp.listDirectories("/home/hadoop");
			for(FTPFile f:ftpDir){
				//System.out.println("test the ftp server directory:" + f.getName());
			}*/
			//将上传文件存储到指定目录
			String fileName = "";
			String gzFileName="";
			try {
				Path newFilePath = new Path(successHdfsPath);
				if((!HdfsUtil.existsHdfsPath(newFilePath))||(!HdfsUtil.isHdfsDirectory(newFilePath))){
					//如果,待放上传成功的文件的目录不存在,则创建一个新目录
					boolean mdRet=HdfsUtil.makeHdfsDir(newFilePath);
					log.debug("Create the sucess hdfs directory :" + mdRet);
				}

				for( Path file:files){
					if (file==null){
						continue;
					}
					if (HdfsUtil.isHdfsFile(file)) {
						// 遍历待上传的文件
						fileName = file.getName();
						if(filePostfix!=null){
							if(!filePostfix.equals(".*")){
								if(!fileName.endsWith(filePostfix)){
									log.info("Not a specified file : " + fileName);
									return true;
								}
							}
						}
						// 上传文件
						log.info("Begin upload file :" + fileName);
						if(isGZip){
//								将文件拆分
							//按行读取输入文件
							FSDataInputStream in=HdfsUtil.ReadHDFSFileStream(file);
							BufferedReader lineRead=new BufferedReader(new InputStreamReader(in,"UTF-8"));
							//通过路径获得文件类型，判断文件输出类型（LUA或NUA）
							String[] fileParent=file.getParent().toString().split("/");
							//文件类型索引改为可配置
							//String fileType=fileParent[5].substring(1);
							String fileType=fileParent[fileTypeIndex];
							String fileNamePre="";
							if(fileType.contains("NetPub")){
								fileNamePre="LUA";
							}else{
								fileNamePre="NUA";
							}

							String line="";
							int temp =0;
							SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");

							//log.info("FTPUtil...Uploading the gzFile:"+gzFileName);
							//SXS:file catch****************************************
							ArrayList<byte[]> fileContent=new ArrayList<byte[]>();
							String id="";
							while((line=lineRead.readLine())!=null){
								temp++;
								line+="\r\n";
								fileContent.add(line.getBytes());
								if(temp==lineLimit){
									id= FileIdGenerator.getId();
									gzFileName=fileNamePre +"_" + id + "_"+ sdf.format(new Date()) + ".xml.gz";
									uploadGzFile(ftp, gzFileName, id, fileContent, temp);
									fileContent.clear();
									temp=0;
								}
							}
							if(temp!=0){
								id= FileIdGenerator.getId();
								gzFileName=fileNamePre +"_" + id + "_"+ sdf.format(new Date()) + ".xml.gz";
								uploadGzFile(ftp, gzFileName, id, fileContent, temp);
								fileContent.clear();
							}
						}
					}

					// 将上传成功的文件转移到新目录中
					log.debug("==============begin move file ========" + fileName);
					//log.info("==============begin move file ========" + fileName);
					HdfsUtil.renameHdfsFile(file, new Path(successHdfsPath + "/" + file.getName()));
					//log.info("==============end move file ========" + fileName);
					//log.debug("==============end move file ========"+fileName);
					success = true;
				}
			} catch (Exception e) {
				e.printStackTrace();
				log.error("上传文件["+fileName+"|"+gzFileName+"]中的文件出错!");
				throw new Exception("上传文件["+fileName+"|"+gzFileName+"]中的文件出错!");
			}

		} catch (Exception e1) {
			e1.printStackTrace();
		} finally{
			try {
				if(ftp.isConnected()){
					ftp.logout();
					ftp.disconnect();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		//退出FTP
		return success;
	}
	private static void uploadGzFile (FTPClient ftp, String newName, String id, ArrayList<byte[]> content, int lineNum ) throws Exception{
		log.info("Uploading the gzFile:"+newName);
		OutputStream out = ftp.appendFileStream(newName);
		//add the xml file head
		String head="<?xml version=\"1.0\" encoding=\"UTF-8\"?>\r\n"
				+ "<info id=\""+ id + "\" type=\"nat_cdr_info\" resultnum=\"" + lineNum +"\">\r\n";
		byte[] gzBuffer=GzUtil.gZip(head.getBytes(),head.getBytes().length);
		out.write(gzBuffer, 0, gzBuffer.length);
		//wirte the file content

        for(byte[] buf : content){
            gzBuffer=GzUtil.gZip(buf,buf.length);
            out.write(gzBuffer, 0, gzBuffer.length);
        }
		//add the xml file tail
		String tail="</info>\r\n";
		gzBuffer=GzUtil.gZip(tail.getBytes(),tail.getBytes().length);
		out.write(gzBuffer, 0, gzBuffer.length);
	 	//close the outputstream
		out.close();

    	 // Must call completePendingCommand() to finish command.
    	 if(!ftp.completePendingCommand()) {
    	     ftp.logout();
    	     ftp.disconnect();
    	     throw new Exception("上传文件["+newName+"]中的文件出错!");
		 }

		//reanme after uploading successfully
		boolean ret=ftp.rename(newName, newName+".ok");
		if(!ret) throw new Exception("重命名文件["+newName+"]时出错");
	}
}
